package com.sdses.flink.cep;

import com.ibm.icu.text.SimpleDateFormat;
import com.sdses.flink.sql.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.*;

public class Flink_CEP_Demo01 {
    private static final OutputTag<WaterSensor> WARNING_OUTPUT_TAG  = new OutputTag<WaterSensor>("unknownSensor") {
    };

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        env.setParallelism(1);

        List<String> knownSensorList = new ArrayList<>();
        knownSensorList.add("sensor_1");
        knownSensorList.add("sensor_22");


        SingleOutputStreamOperator<WaterSensor> process = env.addSource(new SensorDataSource())
                .keyBy(WaterSensor::getId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<WaterSensor, WaterSensor, String, TimeWindow>() {
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");


                    @Override
                    public void open(Configuration parameters) throws Exception {

                        System.out.println("窗口初始化....");
                    }

                    @Override
                    public void process(String key, ProcessWindowFunction<WaterSensor, WaterSensor, String, TimeWindow>.Context context, Iterable<WaterSensor> iterable, Collector<WaterSensor> collector) throws Exception {
                        //统计当前窗口内不同key的数量
                        Map<String, Integer> keyCountMap = new HashMap<>();
                        for (WaterSensor sensor : iterable) {
                            String sensorKey = sensor.getId();
                            if (keyCountMap.containsKey(sensorKey)) {
                                keyCountMap.put(sensorKey, keyCountMap.get(sensorKey) + 1);
                            } else {
                                keyCountMap.put(sensorKey, 1);
                            }
                        }
                        String windowStartStr = sdf.format(new Date(context.window().getStart()));
                        String windowEndStr = sdf.format(new Date(context.window().getEnd()));


                        System.out.println("窗口 [" + windowStartStr + ", " + windowEndStr + "] 内不同key的数量统计: " + keyCountMap);
// 检查每个key的出现次数，将出现次数为0的key输出到侧输出流作为告警信息，>0的数据主流输出
                        for (Map.Entry<String, Integer> entry : keyCountMap.entrySet()) {
                            if (entry.getValue() == 0) {
                                System.out.println( "告警：在窗口 [" + context.window().getStart() + ", " + context.window().getEnd() + "] 内，key为 " + entry.getKey() + " 的数据未出现");
//                                context.output(WARNING_OUTPUT_TAG, "告警：在窗口 [" + context.window().getStart() + ", " + context.window().getEnd() + "] 内，key为 " + entry.getKey() + " 的数据未出现");
                            } else {
                                // 将原始数据收集到主流
                                for (WaterSensor sensor : iterable) {
                                    if (sensor.getId().equals(entry.getKey())) {
                                        collector.collect(sensor);
                                    }
                                }
                            }
                        }

                    }
                });

        // 获取侧输出流
        SideOutputDataStream<WaterSensor> warningStream = process.getSideOutput(WARNING_OUTPUT_TAG);

        // 可以对侧输出流和主流分别进行后续处理，这里简单打印输出示例
        process.print("主流输出");
        warningStream.print("侧输出流（告警信息）");


//        SingleOutputStreamOperator<WaterSensor> mainStream = env.addSource(new SensorDataSource())
//                .assignTimestampsAndWatermarks(WatermarkStrategy
//                        .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//                        .withTimestampAssigner((sensor, recordTimestamp) -> sensor.getTs()))
//                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .process(new ProcessAllWindowFunction<WaterSensor, WaterSensor, TimeWindow>() {
//                    @Override
//                    public void process(ProcessAllWindowFunction<WaterSensor, WaterSensor, TimeWindow>.Context context, Iterable<WaterSensor> elements, Collector<WaterSensor> out) throws Exception {
//                        for (WaterSensor element : elements) {
//                            if (knownSensorList.contains(element.getId())) {
//                                out.collect(element);
//                            } else {
//                                context.output(unknownSensorTag, element);
//                            }
//                        }
//                    }
//                });
//
//        DataStream<WaterSensor> unknownSensorStream = mainStream.getSideOutput(unknownSensorTag);
//
//        mainStream.print("Main Stream: ");
//
//        unknownSensorStream.print("Unknown Sensor Stream: ");


        env.execute();

    }

    public static class SensorDataSource implements SourceFunction<WaterSensor> {
        private boolean running = true;
        private Random random = new Random();

        @Override
        public void run(SourceContext<WaterSensor> sourceContext) throws Exception {
            while (running) {
                long currentTime = System.currentTimeMillis();
                // 随机生成传感器id
                String id = "sensor_" + random.nextInt(10);
                // 随机生成传感器值
                Integer vc = random.nextInt(50);
                WaterSensor waterSensor = new WaterSensor(id, currentTime, vc);
                // 将生成的传感器数据发送出去
                sourceContext.collect(waterSensor);
                try {
                    // 每隔一定时间生成一条新数据，这里设置为每秒生成一条（可根据需要调整）
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }


        @Override
        public void cancel() {
            running = false;
        }
    }

}

